package com.example;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Demo1 {
    public static void main(String[] args) throws InterruptedException {

        Flux<Integer> flux = Flux.range(1, 7)
                .doOnComplete(() -> {
                    System.out.println("流正常结束...");
                })
                .doOnCancel(() -> {
                    System.out.println("流已被取消...");
                })
                .doOnError(throwable -> {
                    System.out.println("流出错..." + throwable);
                })
                .doOnNext(integer -> {
                    System.out.println("doOnNext..." + integer);
                }); //有一个信号：此时代表完成信号

      /*  flux.subscribe(
                v -> System.out.println("v = " + v), //流元素消费
                throwable -> System.out.println("throwable = " + throwable), //感知异常结束
                () -> System.out.println("流结束了...") //感知正常结束
        );*/


      /* Flux<Integer> flux = Flux.range(1, 100);

        flux.subscribe(
                v -> System.out.println("v = " + v), //流元素消费
                throwable -> System.out.println("throwable = " + throwable), //感知异常结束
                () -> System.out.println("流结束了...") //感知正常结束
        );

        flux.subscribe(new BaseSubscriber<>() {
            // 生命周期钩子1： 订阅关系绑定的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                // 流被订阅的时候触发
                System.out.println("绑定了..." + subscription);

                //找发布者要数据
                request(1); //要1个数据
                // requestUnbounded(); //要无限数据
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("数据到达，正在处理：" + value);
                try {
                    TimeUnit.MILLISECONDS.sleep(2000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                request(1); //要1个数据
            }

            //  hookOnComplete、hookOnError 二选一执行
            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束...");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常..." + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消...");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调...一定会被执行");
            }
        });*/
    }
}
